package com.zyb.dfs.datanode.server;

import java.io.File;
import java.io.FileOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.ConcurrentLinkedQueue;

import static com.zyb.dfs.datanode.server.DataNodeConfig.DATA_DIR;

import com.alibaba.fastjson.JSONObject;

/**
 * 副本复制管理组件
 *
 * @author dhh
 */
public class ReplicateManager {
    public static final Integer REPLICATE_THREAD_NUM = 3;

    /**
     * NIOClient
     */
    private NIOClient nioClient = new NIOClient();

    // 与NameNode进行通信的客户端
    private NameNodeRpcClient namenodeRpcClient;

    /**
     * 副本复制任务队列
     */
    private ConcurrentLinkedQueue<JSONObject> replicateTaskQueue =
            new ConcurrentLinkedQueue<JSONObject>();

    public ReplicateManager(NameNodeRpcClient namenodeRpcClient) {
        this.namenodeRpcClient = namenodeRpcClient;
        for (int i = 0; i < REPLICATE_THREAD_NUM; i++) {
            new ReplicateWorker().start();
        }
    }

    /**
     * 添加复制任务
     *
     * @param replicateTask
     */
    public void addReplicateTask(JSONObject replicateTask) {
        replicateTaskQueue.offer(replicateTask);
    }

    /**
     * 副本复制线程
     *
     * @author zhonghuashishan
     */
    class ReplicateWorker extends Thread {

        @Override
        public void run() {
            while (true) {
                FileOutputStream imageOut = null;
                FileChannel imageChannel = null;

                try {
                    JSONObject replicateTask = replicateTaskQueue.poll();
                    if (replicateTask == null) {
                        Thread.sleep(1000);
                        continue;
                    }

                    // 解析复制任务
                    String filename = replicateTask.getString("filename");

                    JSONObject sourceDatanode = replicateTask.getJSONObject("sourceDatanode");
                    String hostname = sourceDatanode.getString("hostname");
                    Integer nioPort = sourceDatanode.getInteger("nioPort");

                    // 跟源头数据接头通信读取图片过来
                    byte[] file = nioClient.readFile(hostname, nioPort, filename);
                    ByteBuffer fileBuffer = ByteBuffer.wrap(file);

                    // 根据文件的相对路径定位到绝对路径，写入本地磁盘文件中
                    String absoluteFilename = getAbsoluteFilename(filename);

                    imageOut = new FileOutputStream(absoluteFilename);
                    imageChannel = imageOut.getChannel();

                    imageChannel.write(fileBuffer);
                    // 增量上报
                    namenodeRpcClient.informReplicaReceived(filename);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    try {
                        if (imageChannel != null) {
                            imageChannel.close();
                        }
                        if (imageOut != null) {
                            imageOut.close();
                        }
                    } catch (Exception e2) {
                        e2.printStackTrace();
                    }
                }
            }
        }

        /**
         * 获取文件在本地磁盘上的绝对路径名
         *
         * @param relativeFilename
         * @return
         * @throws Exception
         */
        private String getAbsoluteFilename(String relativeFilename) throws Exception {
            String[] relativeFilenameSplited = relativeFilename.split("/");

            String dirPath = DATA_DIR;
            for (int i = 0; i < relativeFilenameSplited.length - 1; i++) {
                if (i == 0) {
                    continue;
                }
                dirPath += "\\" + relativeFilenameSplited[i];
            }

            File dir = new File(dirPath);
            if (!dir.exists()) {
                dir.mkdirs();
            }

            String absoluteFilename = dirPath + "\\" +
                    relativeFilenameSplited[relativeFilenameSplited.length - 1];
            return absoluteFilename;
        }

    }

}
